package com.monkeyboy.demo.base.consumer;

import lombok.SneakyThrows;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * 消息消费者
 *
 * @Author Gavin
 * @date 2020.08.18 12:54
 */
public class Consumer {
    static volatile boolean flag = false;

    public static void main(String[] args) throws Exception {
//        1.创建消费者Consumer，制定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group");
//        2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.1.140:9876;192.168.1.141:9876");
//        3.订阅主题Topic和Tag(第二个参数就是tag,支持*)
        consumer.subscribe("test", "*");
        //设定消费模式：负载均衡(默认)|广播模式
//        consumer.setMessageModel(MessageModel.CLUSTERING);
        //默认的消费位置默认是CONSUME_FROM_LAST_OFFSET，最新位置开始消费
//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
//        4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //接受消息
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println(new String(msg.getBody()));
                    if(msgs.get(0).getReconsumeTimes()>0){
                        System.out.println("重试次数：  "+msgs.get(0).getReconsumeTimes());
                    }
//                   Math.ceil(Math.random()*10)%2==0
                    if (!flag) {
                        System.out.println(new String(msg.getBody()) + "  稍后会进行重试。");
                        flag=true;
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//消费成功或者失败后要给一个反馈
            }
        });
//        5.启动消费者consumer
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }

}
